package cn.dmrliu.edu.realtime.app.func;


import cn.dmrliu.edu.realtime.bean.TableProcess;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

public class DwdTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Map<String, TableProcess> mapConfig = new HashMap<>();

    public DwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.cj.jdbc.Driver");

        String url = "jdbc:mysql://h102:3306/edu_config";

        Properties properties = new Properties();
        properties.put("user", "root");
        properties.put("password", "j)k*x~y0)*n_L)!o{y?C`w(1rV_viw");
        properties.put("useUnicode", "true");
        properties.put("characterEncoding", "utf8");
        properties.put("serverTimeZone", "Asia/Shanghai");
        properties.put("useSSL", "false");

        String sql = "select * from edu_config.table_process where sink_type = 'dwd'";

        Connection connection = DriverManager.getConnection(url, properties);
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();

        while (resultSet.next()) {
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = resultSet.getObject(i);
                jsonObject.put(columnName, columnValue);
            }
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            mapConfig.put(tableProcess.getSourceTable(), tableProcess);
        }

        resultSet.close();
        preparedStatement.close();
        connection.close();
    }

    // 主流数据处理
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // TODO 1.获取表,算子状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = null;
        // 拼接配置表的key
        String sourceTable = jsonObject.getString("table");
        String sourceType = jsonObject.getString("type");
        String key = sourceTable + ":" + sourceType;

        // TODO 2.筛选维度信息处理
        if ((tableProcess = broadcastState.get(key)) != null
                || (tableProcess = mapConfig.get(key)) != null) {
            // 获取数据信息
            JSONObject dataJSONObject = jsonObject.getJSONObject("data");
            // 获得列信息
            String sinkColumns = tableProcess.getSinkColumns();
            // 移除不需要的列
            filterColumn(dataJSONObject, sinkColumns);
            // 获取写入的表信息
            String sinkTable = tableProcess.getSinkTable();
            // 补充信息
            dataJSONObject.put("sink_table", sinkTable);
            out.collect(dataJSONObject);
        }
    }

    /**
     * 清除不用的列
     *
     * @param dataJSONObject
     * @param sinkColumns
     */
    private void filterColumn(JSONObject dataJSONObject, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJSONObject.entrySet();
        entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
    }


    // 配置流数据处理
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        // TODO 1.json字符串转为json对象
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);

        // TODO 2.根据操作类型执行操作
        String op = jsonObject.getString("op");//获取操作符
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);//获取广播状态
        // 广播状态中保存配置信息，配置信息的sourceTable:sourceType作为key
        if ("d".equals(op)) {
            TableProcess before = jsonObject.getObject("before", TableProcess.class);
            String sinkType = before.getSinkType();
            if ("dwd".equals(sinkType)) {
                // 拼接key
                String sourceTable = before.getSourceTable();
                String sourceType = before.getSourceType();
                String key = sourceTable + ":" + sourceType;
                broadcastState.remove(key);
                mapConfig.remove(key);
            }
        } else {
            TableProcess after = jsonObject.getObject("after", TableProcess.class);
            String sinkType = after.getSinkType();
            if ("dwd".equals(sinkType)) {
                // 拼接key
                String sourceTable = after.getSourceTable();
                String sourceType = after.getSourceType();
                String key = sourceTable + ":" + sourceType;
                broadcastState.put(key, after);
//                System.out.println("配置信息>>>>>" + after);
                mapConfig.put(key, after);
            }
        }
    }


}
